<?php
/* vim: set expandtab tabstop=4 shiftwidth=4 encoding=utf-8: */
/* $Id: */
/*
Copyright (c) 2010, Justin Swanhart
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
    * Redistributions of source code must retain the above copyright
      notice, this list of conditions and the following disclaimer.
    * Redistributions in binary form must reproduce the above copyright
      notice, this list of conditions and the following disclaimer in the
      documentation and/or other materials provided with the distribution.
    * Neither the name of the <organization> nor the
      names of its contributors may be used to endorse or promote products
      derived from this software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY
DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/

/*This script requires PEAR Net_Gearman */
/*It also requires Console_Getopt, but this should be installed by default with pear */
require_once 'shard-query.php';

#FIXME: This should not extend MySQLDAl, but should create a new DL object
#DL needs additions such as creating table, check existence of table, etc.
#This will be useful for FlexCDC and Shard-Query.
class ShardLoader extends ShardQuery {

	public function delimited_loader($path,$table,$loadspec,$columns, $use_fifo=true, $start=0, $end=0,$line_terminator="\r\n") {
		if(!trim($path)) return false; 

		#this is the shard where non-partitioned records go.
		#this is probably the directory

		if($this->verbose) {
			echo "== delimited loader invoked ==\n";
			echo "== file: $path\n";
			echo "== table: $table ==\n";
			echo "== partition column: " . ($this->partition_column ? $this->partition_column : 'NULL') . "\n";
			echo "== loader spec  ==\n";
			echo print_r($loadspec, true);
			echo "== columns ==\n";
			echo print_r($columns,true);
	
		}

		if(!is_array($columns)) {
			echo "List of columns is expected to be an array!\n";
			return false;
		}

		$ignore = false;
		$errors=false;

		if(!empty($loadspec['ignore'])) $ignore=$loadspec['ignore'];

		if($path != '-') {
			if(!$fh = fopen($path,'r')) throw new Exception('could not open input stream\n');
		} else {
			if(!$fh = fopen('php://stdin', 'r')) throw new Exception('could not open input stream\n');
		}
		
		if(!$fh) throw new Exception('Could not open ' . $path . " for reading\n");
		$shard_col_pos = false;

		if($this->partition_column ) {
			foreach($columns as $key => $column) {
				if($column == $this->partition_column) {
					$shard_col_pos = $key;
					break;
				}
			}
		}

		if($shard_col_pos === false && $this->partition_column) {
			if($this->verbose) {
				echo("== Did not find column: '{$this->partition_column}' in list of columns: ");
				echo("rows will be sent to all shards\n"); 
			}
		}
	
		if(!$this->partition_column) {
			if($this->verbose) {
				echo("== Partition column is not set: ");
				echo("rows will be sent to all shards\n"); 
			}
		}

		$loadspec['delimiter'] = str_replace(array('\\','|','/'),array("\\","\\|", "\\/"), $loadspec['delimiter']);

		if(!empty($loadspec['enclosure'])) {
			$regex = "/{$loadspec['enclosure']}([^{$loadspec['enclosure']}]*){$loadspec['enclosure']}|{$loadspec['delimiter']}/ix";
		} else {
			$regex = "/{$loadspec['delimiter']}/";
		}

		if($this->verbose) {
			echo "== regex from loadspec: $regex\n";
		}

		#read each line from the file, then split it up and look up the shard based on the value
		#the mapper is expected to create values if they do not already exist in the mapping

		$loader_handles = array();
		$loader_fifos = array();

		$errors = false;
		$line_num = 0;	
		$columns_str = join(",", $columns);
		$skipped=false; #have we skipped the number of lines we were asked to skip?


		while(!feof($fh)) {
			if($end >0) {
				$lines = ShardQuery::read_chunk($fh, $start, $end, $line_terminator,10000,$this->chunk) ;
				#we only skip forward once
				$start=false;
			} else {
				$skipped=true; #skipping only happens in the first chunk
				$lines = fgets($fh);
				$lines = array($lines);
			}

			foreach($lines as $line) {
				if(!trim($line)) continue;
				#echo "[{$this->chunk}]: $line\n";
				++$line_num;
				if(!$ignore && $errors !== false) {
					print_r($errors);
				#	break;
				}
				if(!$skipped && !empty($loadspec['skip']) && $loadspec['skip'] >= $line_num)  continue;
				
				$values = array();

				$this->force_shard=false;
				$info=false;
				if($shard_col_pos !== false) {
					
                			$values = preg_split($regex, $line,-1,PREG_SPLIT_NO_EMPTY| PREG_SPLIT_DELIM_CAPTURE );
					$info=$this->get_partition_info($this->partition_column, $values[$shard_col_pos]);

					if(empty($values[$shard_col_pos])) continue;
					$info=$this->get_partition_info($this->partition_column, $values[$shard_col_pos]);
					
					if(!$info) {
						$err = "Discarded row because the partition mapper did not return a valid result for the given partition column.\n";
						$errors[] = array($err=> $line);
						continue;
					} 
					if(is_array($info)) {
						$keys=array_keys($info);
						$this->force_shard=$keys[0];
					} else {
						$this->force_shard=$info;
					}
				}
				
				#we need a list of values for do_insert, which we only use if use_fifo is false
				if(!$use_fifo && empty($values)) {
                			$values = preg_split($regex, trim($line),-1,PREG_SPLIT_NO_EMPTY| PREG_SPLIT_DELIM_CAPTURE );
				}

				foreach($this->shards as $shard_key => $shard) {
					#only send the row the correct shard
					if($this->force_shard && ($this->force_shard != $shard_key)) {
						continue;
					} 
					if($use_fifo) {
						if(empty($loader_fifos[$shard_key]))	{
							if(!empty($loadspec['method'])) {
								switch(strtolower($loadspec['method'])) {
									case 'split':
										$fifo = $this->start_split($table,$loadspec,$shard);
										break;
		
									default:
										$fifo = $this->start_fifo($table,$loadspec,$shard,$columns_str);
								}

							} else {
								$fifo = $this->start_fifo($table,$loadspec,$shard,$columns_str);
							}
							if($fifo == false) {
								return(array(array("Could not open FIFO for table:{$table}" => "for shard:" . print_r($shard, true))));
							} 
							$loader_fifos[$shard_key] = $fifo;
						}
						#FIXME: check how much data was written
						$result = fwrite($loader_fifos[$shard_key]['fh'], $line);
					} else 	{
						if(empty($loader_handles[$shard_key]))	{
							
							$dal = $loader_handles[$shard_key] = SimpleDAL::factory($shard);
							$dal->my_query('begin');

						}
						$result = $this->do_insert($values,$loadspec,$table,$columns,$columns_str,$loader_handles,$shard_key); 
						if(is_array($result)) {	
							print_r($result);
							// exit;
						} else {
							if($result === false) {
								throw new Exception('Could not insert!\n');
							}
						}
					}
				

					if(empty($loader_handles[$shard_key]) && empty($loader_fifos[$shard_key])) {
						$err= "Discarded row because the destination database could not be connected.\n";
						$errors[] =array($err=> $line);
						continue;

					}
				}
			}
			if($end && ftell($fh) >= $end)  break;
		}				

		foreach($loader_handles as $handle) {
			$handle->my_query('commit');
			$handle->my_close();
		}
		
		#this will automatically end the process reading from the other end of the named pipe

		foreach($loader_fifos as $handle) {
			#fputs($handle['fh'], 0x04);
			#if(feof($fh)) fputs($handle['fh'],"\n");
			fflush($handle['fh']);
			fclose($handle['fh']);
		}


		$this->errors = $errors;

		return($this->errors);
		
	}


	#used by the loader to insert a row
	protected function do_insert(&$values,&$loadspec,&$table,$columns,&$columns_str,&$loader_handles,$shard_key) {
		$vals = "";
	
		foreach($columns as $colkey => $column ) {
			if(empty($values[$colkey])) $values[$colkey] = '';
		}
	
		$vals = "'" . join("','", $values) . "'";
		if(empty($loadspec['ignore'])) $loadspec['ignore'] = ''; else $loadspec['ignore']='IGNORE';
		$sql = "INSERT {$loadspec['ignore']} INTO $table ($columns_str) VALUES ($vals)"; 

		$dal = $loader_handles[$shard_key];	
		if(!$dal->my_query($sql)) {
			$err = "WARNING: Discarded row because the insertion into destination database failed:$sql\n";
			return array($err=>$dal->my_error());
		}

		return true;
	}

	protected function start_split($table,$loadspec,$shard) {

		$fname =  str_replace('.','',microtime(true)) . '.txt';

		if(empty($loadspec['output_directory'])) $loadspec['output_directory'] = './';
		$path = $loadspec['output_directory'];
		if(substr($path,-1) != '/') {
			$path .= "/";
		}
		$path .= $shard['host'] . '/' . $shard['db'] . '/' . $table . '/';
		if(!file_exists($path)) if(!mkdir($path,0777,true)) throw new Exception("could not create $path\n");
		$path .= $fname;

		if(!$fh = fopen($path, 'wb')) throw new Exception('Could not open ' . $path . " for writing\n");	

		return(array('fh' => $fh, 'ph' => null)); 

	}


	protected function start_fifo($table,$loadspec,$shard, $columns_str) {

		$rand = mt_rand(1,100000000);
		$path = '/tmp/sq_tmp_fifo#' . $rand;
		$result = posix_mkfifo($path,'0700');
		if(!$result) {
			return array("Could not open FIFO" => $path);
		}

		if(empty($loadspec['ignore'])) $loadspec['ignore'] = ' '; else $loadspec['ignore'] = ' IGNORE ';
		$sql = "";

		$load="LOAD DATA LOCAL INFILE \"$path\" IGNORE INTO TABLE `{$shard['db']}`.`$table`";
		if(!empty($loadspec['delimiter'])) {
			$sql = " FIELDS TERMINATED BY \"" . $loadspec['delimiter'] . "\" ";
		}

		if(!empty($loadspec['enclosure'])) {
			if($sql == "") {
				 $sql = ' FIELDS ';
			}
			
			$sql .= "OPTIONALLY ENCLOSED BY \"" . $loadspec['enclosure'] . "\" ";
			
		}
		if(empty($loadspec['line_terminator'])) $loadspec['line_terminator'] = "\\n";
		$line_terminator = str_replace(array("\r","\n"),array("\\r","\\n"), $loadspec['line_terminator']);

		$sql .= "LINES TERMINATED BY \"" . $line_terminator . "\" ";
		#$sql .= "LINES TERMINATED BY \"\\n\" ";

		if(!empty($loadspec['skip'])) {
			$sql .= "SKIP {$loadspec['skip']} LINES ";
		}

		$sql .= "($columns_str)";

		$load .= $sql;

		$cmdline = "mysql -u{$shard['user']} -h{$shard['host']} -P{$shard['port']} {$shard['db']}";
		if(!empty($shard['password']) && trim($shard['password'])) {
			$cmdline .= " -p{$shard['password']}";
		}
		$cmdline .= " -e 'set @bh_dataformat=\"variable\";{$load}'";
		
		$pipes=null;	
		echo $cmdline . "\n";
		$ph = proc_open($cmdline, array(), $pipes);

		if(!$ph) {
			return array("Could not open FIFO for table {$table}, for shard"=>print_r($shard,true));
		}

		#FIXME
		#we need to check to see if there was an error running the mysql client

		#this won't block because we've started a reader in a coprocess 
		$fh = fopen($path, 'w');	

		return(array('fh' => $fh, 'ph' => array('ph'=>$ph, 'pipes'=>$pipes)));

	}


	protected static function read_chunk($fh, $start, $end, $line_terminator = "\r\n",$batch_size=10000,$chunk=0) {
		$lines = array();
		$fgets_safe=false; #this is only safe for terminators like \r\n or \n.  fgets() can be called, which is 8x faster
		
		#When we start reading at offset zero, then we must not skip
		#to the first line terminator and instead must read all chars.
		#set $started = true to make sure this happens.
		if($start == 0) {
			$started = true;	
		} else {
			$started = false;
		}
	
		#seek to the requested starting position (which might be zero)
		if($start >0) {
			fseek($fh, $start, SEEK_SET);
		}
	
		#this var holds the current line that is being built.  It only
		#can be considered an entire line once a line terminator has
		#been located or we hit eof
		$line = "";
	
		#stop reading when we have reached the end of file
		$terminator_len = strlen($line_terminator);

		if(substr($line_terminator,-1,1) == "\n") {
			$fgets_safe=1;
		}

		while(!feof($fh)) {

			/* Most of the time the delimiter will be only one char (newline)
			but sometimes it might more more than one.  Always read at least
			the number of bytes in the delimiter */
			if($fgets_safe==1) {
				$chars = fgets($fh);
			} else {
				$chars = fread($fh, $terminator_len);
			}

			$last_chars = substr($chars,-1, $terminator_len);

			/* Skip all input until we encountered a line terminator,
	                then start reading.  The only time that $started will
			be true at the start of the loop is when the requested
			starting offset is zero (see above).
			*/
	
			if(!$started) {
				if($last_chars == $line_terminator) {
					$started = true;
				}
				continue;
			}
	
		
			#If chars contains the end-of-the-line
			if($last_chars == $line_terminator) {
				$lines[]=$line . $chars;
				$line = "";

				if( $batch_size !== false && count($lines) >= $batch_size) {
					break;
				}

				if(ftell($fh) >= $end) {
					break; 
				}	
				
			} else {
				$line .= $chars;
			}

		}
	
		if(feof($fh) && trim($line)) {
			$lines[]=$line; 
			$line = "";
		}

		return $lines;	
	}
}
